package com.wang.tool.invoker;


import com.wang.tool.monitor.InvokeTemplate;
import com.wang.tool.monitor.LinkLogUtil;
import com.wang.tool.monitor.core.MonitorInfo;
import com.wang.tool.monitor.template.MonitorInvoker;
import com.wang.tool.monitor.translator.GlobalErrorTranslator;
import com.wang.tool.util.CommonUtil;
import com.wang.tool.common.ErrorInfo;
import com.wang.tool.common.Pair;
import com.wang.tool.common.SimpleException;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 注意:线程池尽量不要是丢弃策略,在拥堵时会导致等到超时才返回(如果无超时设置则会导致死等)
 *
 * @Description 并发执行器
 * @Author shaopeng
 * @Date 2021/6/17
 * @Version 1.0
 */
public class ConcurrentInvoker {


    public static final String TAG_RUN_ERROR_IGNORE = "RunErrorIgnore";
    public static final String TAG_CONCURRENT_TAKE_RESULT = "ConcurrentTakeResult";

    /**
     * 异步批量处理
     *
     * @param action       业务码 建议非空
     * @param params       总参数
     * @param pageSize     单次数量
     * @param processor    处理逻辑
     * @param <P>          入参
     * @param errorHandler 异常处理器,为空时遇到异常直接抛出,int=-1 等待结果时失败, int>=0 出错的任务节点下标 errorHandler中抛出的异常会抛出到主线程
     * @return
     */
    public static <P> void batchPageProcess(String action, ExecutorService executor, Integer timeout, List<P> params, int pageSize, Consumer<List<P>> processor, BiConsumer<Integer, Throwable> errorHandler) {
        if (executor == null) {
            throw new IllegalArgumentException("executor can not be null");
        }
        if (pageSize < 1) {
            throw new IllegalArgumentException("pageSize can not small 1");
        }
        if (processor == null) {
            throw new IllegalArgumentException("processor can not be null");
        }
        if (CommonUtil.isEmpty(params)) {
            return;
        }
        int totalSize = params.size();
        //总的次数
        int allTime = (totalSize + pageSize - 1) / pageSize;
        List<Runnable> runs = new ArrayList<>(allTime);
        for (int i = 0; i < allTime; i++) {
            int toIndex = Math.min((i + 1) * pageSize, totalSize);
            List<P> oneParams = params.subList(i * pageSize, toIndex);
            runs.add(() -> processor.accept(oneParams));
        }
        ConcurrentInvoker.batchRun(action, executor, timeout, runs, errorHandler);
    }

    /**
     * 异步并行处理
     *
     * @param action       业务码 建议非空
     * @param params       总参数
     * @param timeout      超时时间 null=不限制
     * @param processor    处理逻辑
     * @param errorHandler 异常处理器,为空时遇到异常直接抛出,int=-1 等待结果时失败, int>=0 出错的任务节点下标 errorHandler中抛出的异常会抛出到主线程
     * @param <P>          入参
     * @return
     */
    public static <P> void batchProcess(String action, ExecutorService executor, Integer timeout, Collection<P> params, Consumer<P> processor, BiConsumer<Integer, Throwable> errorHandler) {
        if (executor == null) {
            throw new IllegalArgumentException("executor can not be null");
        }
        if (processor == null) {
            throw new IllegalArgumentException("processor can not be null");
        }
        if (CommonUtil.isEmpty(params)) {
            return;
        }
        List<Runnable> runs = new ArrayList<>(params.size());
        for (P param : params) {
            runs.add(() -> processor.accept(param));
        }
        ConcurrentInvoker.batchRun(action, executor, timeout, runs, errorHandler);
    }


    /**
     * 并行执行
     * 带监控
     *
     * @param action       业务码 建议非空
     * @param executor     线程池 线程池的拒绝策略千万不要是丢弃在非超时场景会出现死等情况
     * @param errorHandler 异常处理器,为空时遇到异常直接抛出,int=-1 等待结果时失败, int>=0 出错的任务节点下标 errorHandler中抛出的异常会抛出到主线程
     * @param timeout      等待时间,ms,为空代表死等,强烈建议设置
     * @param runs         需要执行的任务列表
     */
    public static void batchRun(String action, ExecutorService executor, Integer timeout, List<Runnable> runs, BiConsumer<Integer, Throwable> errorHandler) {
        if (executor == null) {
            throw new IllegalArgumentException("executor can not be null");
        }
        if (timeout != null && timeout <= 0) {
            throw new IllegalArgumentException("timeout can not smaller 1");
        }
        if (CommonUtil.isEmpty(runs)) {
            return;
        }

        //开始时间
        long begin = System.currentTimeMillis();
        //成功计数器
        AtomicInteger successNum = new AtomicInteger();
        /**
         * 监控执行
         */
        InvokeTemplate.monitor("concurrent", "batchRun", action, new MonitorInvoker<Void>() {
            @Override
            public void weaveOnBefore(MonitorInfo<Void> monitorInfo) {
                monitorInfo.addDigest("allNum", runs.size());
                monitorInfo.addDigest("timeout", timeout);
            }

            @Override
            public void weaveOnAfter(MonitorInfo<Void> monitorInfo) {
                monitorInfo.addDigest("successNum", successNum.get());
            }

            @Override
            public Void internalCall() throws Throwable {
                //ExecutorCompletionService
                //1.解决普通future无法合并取值的性能问题(挨个future.get()上线文切换性能差)及无法感知先后顺序
                //2.CompletableFuture 提交的future无法阻断调用(cancel(ture)阻断参数无效)
                ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(executor, new LinkedBlockingQueue<>(runs.size()));
                List<Future<Void>> futures = new ArrayList<>(runs.size());
                try {
                    for (int i = 0; i < runs.size(); i++) {
                        Runnable run = runs.get(i);
                        int y = i;
                        try {
                            futures.add(completionService.submit(() -> {
                                try {
                                    run.run();
                                    successNum.incrementAndGet();
                                } catch (Throwable e) {
                                    if (errorHandler == null) {
                                        throw e;
                                    }
                                    errorHandler.accept(y, e);
                                    ErrorInfo error = GlobalErrorTranslator.translate(e);
                                    LinkLogUtil.newTagLog(TAG_RUN_ERROR_IGNORE)
                                            .v1(action)
                                            .v2(error.getErrorCode())
                                            .addAttachment(MonitorInfo.ERROR, error)
                                            .log();
                                }
                            }, null));
                        } catch (Throwable e) {
                            if (errorHandler == null) {
                                throw e;
                            }
                            errorHandler.accept(y, e);
                            ErrorInfo error = GlobalErrorTranslator.translate(e);
                            LinkLogUtil.newTagLog(TAG_RUN_ERROR_IGNORE)
                                    .v1(action)
                                    .v2(error.getErrorCode())
                                    .addAttachment(MonitorInfo.ERROR, error)
                                    .log();
                        }
                    }

                    if (timeout == null) {
                        //死等
                        for (int i = 0; i < futures.size(); i++) {
                            completionService.take();
                        }
                        return null;
                    }
                    //超时阈值
                    long threshold = begin + timeout;
                    //带超时等
                    for (int i = 0; i < futures.size(); i++) {
                        //wait可<0 内部实现和0一致
                        long wait = threshold - System.currentTimeMillis();
                        //按照超时时间取值,时间过了返回null
                        Future<Void> future = completionService.poll(wait, TimeUnit.MILLISECONDS);
                        if (future == null) {
                            throw new TimeoutException();
                        }
                    }
                    return null;
                } catch (Throwable e) {
                    if (errorHandler == null) {
                        throw e;
                    }
                    errorHandler.accept(-1, e);
                    ErrorInfo error = GlobalErrorTranslator.translate(e);
                    LinkLogUtil.newTagLog(TAG_RUN_ERROR_IGNORE)
                            .v1(action)
                            .v2(error.getErrorCode())
                            .addAttachment(MonitorInfo.ERROR, error)
                            .log();
                } finally {
                    //这时方法已经执行结束了,未完成的任务再执行也没有意义
                    for (Future<Void> future : futures) {
                        if (!future.isDone()) {
                            //1.阻断正在执行的
                            //2.阻断未开始的
                            future.cancel(true);
                        }
                    }
                }
                return null;
            }
        });

    }

    /**
     * 并行执行任务并且根据优先级取优先级最高的正确结果
     * 带监控
     *
     * @param <T>
     * @param action    动作码 尽量非空
     * @param executor  线程池 非空
     * @param timeout   超时时间 空代表不限制
     * @param callables 任务 非空 集合前面的任务优先级更高
     * @return
     */
    public static <T> T takeHigherPriority(String action, ExecutorService executor, Integer timeout, List<Callable<T>> callables) {
        return concurrentAndSelect("takeHigherPriority", action, executor, timeout, callables, futures -> {
            //从前往后取,取到最前面合理的结果
            for (Future<Pair<T, Throwable>> future : futures) {
                if (!future.isDone()) {
                    return null;
                }
                try {
                    Pair<T, Throwable> pair = future.get();
                    if (pair.getLeft() != null) {
                        return pair.getLeft();
                    }
                } catch (Throwable e2) {
                    //已经结束 理论上不存在异常
                    throw new RuntimeException(e2);
                }
            }

            //全部执行完了但是没有合理数据,取第一个异常抛出
            for (Future<Pair<T, Throwable>> future : futures) {
                Throwable e;
                try {
                    e = future.get().getRight();
                } catch (Throwable e2) {
                    //已经结束 理论上不存在异常
                    throw new RuntimeException(e2);
                }
                if (e != null) {
                    if (e instanceof SimpleException) {
                        throw (SimpleException) e;
                    }
                    if (e instanceof RuntimeException) {
                        throw (RuntimeException) e;
                    }
                    if (e instanceof Error) {
                        throw (Error) e;
                    }
                    ErrorInfo error = GlobalErrorTranslator.translate(e);
                    throw new SimpleException(e, error.getErrorCode(), error.getErrorMsg());
                }
            }
            return null;
        });
    }

    /**
     * 并行执行任务并且取第一个返回值的有效结果
     * 带监控
     *
     * @param <T>
     * @param action    动作码 尽量非空
     * @param executor  线程池 非空
     * @param timeout   超时时间 空代表不限制
     * @param callables 任务 非空 集合前面的任务优先级更高
     * @return
     */
    public static <T> T takeFaster(String action, ExecutorService executor, Integer timeout, List<Callable<T>> callables) {
        return concurrentAndSelect("takeFaster", action, executor, timeout, callables, futures -> {
            //取有结果的即可
            boolean allDone = true;
            for (Future<Pair<T, Throwable>> future : futures) {
                if (!future.isDone()) {
                    allDone = false;
                    continue;
                }
                try {
                    Pair<T, Throwable> pair = future.get();
                    if (pair.getLeft() != null) {
                        return pair.getLeft();
                    }
                } catch (Throwable e2) {
                    //已经结束 理论上不存在异常
                    throw new RuntimeException(e2);
                }
            }

            if (!allDone) {
                return null;
            }

            //全部执行完了但是没有合理数据,取第一个异常抛出
            for (Future<Pair<T, Throwable>> future : futures) {
                Throwable e;
                try {
                    e = future.get().getRight();
                } catch (Throwable e2) {
                    //已经结束 理论上不存在异常
                    throw new RuntimeException(e2);
                }
                if (e != null) {
                    if (e instanceof SimpleException) {
                        throw (SimpleException) e;
                    }
                    if (e instanceof RuntimeException) {
                        throw (RuntimeException) e;
                    }
                    if (e instanceof Error) {
                        throw (Error) e;
                    }
                    ErrorInfo error = GlobalErrorTranslator.translate(e);
                    throw new SimpleException(e, error.getErrorCode(), error.getErrorMsg());
                }
            }
            return null;
        });
    }

    /**
     * 并行执行并选择合适的结果
     *
     * @param <T>
     * @param action    动作码 尽量非空
     * @param executor  线程池
     * @param timeout   超时时间 空代表不限制
     * @param callables 任务 非空
     * @param tacker    结果选择器,每次有个节点运行完毕了,都会执行,没有并发问题 注意线程阻塞
     * @return
     */
    public static <T> T concurrentAndSelect(String action, ExecutorService executor, Integer timeout, List<Callable<T>> callables, Function<List<Future<Pair<T, Throwable>>>, T> tacker) {
        return concurrentAndSelect("customized", action, executor, timeout, callables, tacker);
    }

    /**
     * 并行执行并选择合适的结果
     *
     * @param <T>
     * @param action    动作码 尽量非空
     * @param executor  线程池
     * @param timeout   超时时间 空代表不限制
     * @param callables 任务 非空
     * @param taker     结果选择器,每次有个节点运行完毕了,都会执行,没有并发问题  注意线程阻塞
     * @return
     */
    private static <T> T concurrentAndSelect(String biz, String action, ExecutorService executor, Integer timeout, List<Callable<T>> callables, Function<List<Future<Pair<T, Throwable>>>, T> taker) {
        if (executor == null) {
            throw new IllegalArgumentException("executor can not be null");
        }
        if (timeout != null && timeout <= 0) {
            throw new IllegalArgumentException("timeout can not smaller 1");
        }
        if (CommonUtil.isEmpty(callables)) {
            return null;
        }
        //开始时间
        long begin = System.currentTimeMillis();
        //成功下标
        List<Integer> successes = new ArrayList<>();
        List<Integer> errors = new ArrayList<>();
        List<Integer> submitFails = new ArrayList<>();
        return InvokeTemplate.monitor("concurrentAndSelect", biz, action, new MonitorInvoker<T>() {
            @Override
            public void weaveOnBefore(MonitorInfo<T> monitorInfo) {
                monitorInfo.addDigest("allNum", callables.size());
                monitorInfo.addDigest("timeout", timeout);
            }

            @Override
            public void weaveOnAfter(MonitorInfo<T> monitorInfo) {
                monitorInfo.addDigest("submitFails", CommonUtil.join(submitFails, ","));
                monitorInfo.addDigest("successes", CommonUtil.join(successes, ","));
                monitorInfo.addDigest("errors", CommonUtil.join(errors, ","));
            }

            @Override
            public T internalCall() {
                ExecutorCompletionService<Pair<T, Throwable>> completionService = new ExecutorCompletionService<>(executor, new LinkedBlockingQueue<>(callables.size()));
                List<Future<Pair<T, Throwable>>> futures = new ArrayList<>(callables.size());
                try {
                    for (int i = 0; i < callables.size(); i++) {
                        int index = i;
                        Callable<T> callable = callables.get(i);
                        try {
                            futures.add(completionService.submit(() -> {
                                try {
                                    T result = callable.call();
                                    successes.add(index);
                                    return new Pair<>(result, null);
                                } catch (Throwable e) {
                                    errors.add(index);
                                    ErrorInfo error = GlobalErrorTranslator.translate(e);
                                    LinkLogUtil.newTagLog(TAG_RUN_ERROR_IGNORE)
                                            .v1(action)
                                            .v2(error.getErrorCode())
                                            .addAttachment(MonitorInfo.ERROR, error)
                                            .log();
                                    return new Pair<>(null, e);
                                }
                            }));
                        } catch (Throwable e) {
                            submitFails.add(index);
                            ErrorInfo error = GlobalErrorTranslator.translate(e);
                            LinkLogUtil.newTagLog(TAG_RUN_ERROR_IGNORE)
                                    .v1(action)
                                    .v2(error.getErrorCode())
                                    .addAttachment(MonitorInfo.ERROR, error)
                                    .log();
                        }
                    }

                    if (timeout == null) {
                        //死等
                        for (int i = 0; i < futures.size(); i++) {
                            Pair<T, Throwable> pair = completionService.take().get();
                            //寻找合适的结果
                            T result = null;
                            Throwable e = null;
                            try {
                                result = taker.apply(futures);
                                if (result != null) {
                                    return result;
                                }
                            } catch (Throwable e2) {
                                e = e2;
                                throw e;
                            } finally {
                                LinkLogUtil.newTagLog(TAG_CONCURRENT_TAKE_RESULT)
                                        .v1(action + "_" + (i + 1) + "_" + (pair.getLeft() != null ? "hasData" : pair.getRight() == null ? "NULL" : pair.getRight().getClass().getSimpleName()))
                                        .v2((result != null ? "hasResult" : "noResult") + (e != null ? ",hasError" : ",noError"))
                                        .log();
                            }
                        }
                        return null;
                    }

                    //超时阈值
                    long threshold = begin + timeout;
                    //带超时等
                    for (int i = 0; i < futures.size(); i++) {
                        //wait可<0 内部实现和0一致
                        long wait = threshold - System.currentTimeMillis();
                        //按照超时时间取值,时间过了返回null
                        Future<Pair<T, Throwable>> future = completionService.poll(wait, TimeUnit.MILLISECONDS);
                        if (future == null) {
                            //超时了,把没有结束的删掉再尝试一次,尽量有可用结果
                            List<Future<Pair<T, Throwable>>> doneFutures = futures.stream().filter(Future::isDone).collect(Collectors.toList());
                            if (!doneFutures.isEmpty()) {
                                T result = null;
                                Throwable e = null;
                                try {
                                    result = taker.apply(doneFutures);
                                    if (result != null) {
                                        return result;
                                    }
                                } catch (Throwable e2) {
                                    e = e2;
                                    throw e;
                                } finally {
                                    LinkLogUtil.newTagLog(TAG_CONCURRENT_TAKE_RESULT)
                                            .v1(action + "_" + (i + 1) + "_timeout")
                                            .v2((result != null ? "hasResult" : "noResult") + (e != null ? ",hasError" : ",noError"))
                                            .log();
                                }
                            }
                            LinkLogUtil.newTagLog(TAG_RUN_ERROR_IGNORE)
                                    .v1(action)
                                    .v2(SimpleException.TIME_OUT)
                                    .log();
                            return null;
                        }

                        Pair<T, Throwable> pair = future.get();
                        T result = null;
                        Throwable e = null;
                        try {
                            result = taker.apply(futures);
                            if (result != null) {
                                return result;
                            }
                        } catch (Throwable e2) {
                            e = e2;
                            throw e;
                        } finally {
                            LinkLogUtil.newTagLog(TAG_CONCURRENT_TAKE_RESULT)
                                    .v1(action + "_" + (i + 1) + "_" + (pair.getLeft() != null ? "hasData" : pair.getRight() == null ? "NULL" : pair.getRight().getClass().getSimpleName()))
                                    .v2((result != null ? "hasResult" : "noResult") + (e != null ? ",hasError" : ",noError"))
                                    .log();

                        }
                    }
                    return null;
                } catch (Throwable e) {
                    ErrorInfo error = GlobalErrorTranslator.translate(e);
                    LinkLogUtil.newTagLog(TAG_RUN_ERROR_IGNORE)
                            .v1(action)
                            .v2(error.getErrorCode())
                            .addAttachment(MonitorInfo.ERROR, error)
                            .log();
                    return null;
                } finally {
                    //这时方法已经执行结束了,未完成的任务再执行也没有意义
                    for (Future<Pair<T, Throwable>> future : futures) {
                        if (!future.isDone()) {
                            //1.阻断正在执行的
                            //2.阻断未开始的
                            future.cancel(true);
                        }
                    }
                }
            }
        });
    }
}
